package com.kafka.comsumer.config;

import com.rabbitmq.entity.User;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.MethodParameter;
import org.springframework.kafka.annotation.KafkaListenerConfigurer;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerEndpointRegistrar;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.kafka.support.SimpleKafkaHeaderMapper;
import org.springframework.kafka.support.converter.MessagingMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.converter.MappingJackson2MessageConverter;
import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

/**
 * @作者: 林江
 * @创建时间: 2024/4/2
 * @功能:
 */
@Configuration
//@EnableKafka
public class KafkaConsumerConfig implements KafkaListenerConfigurer {

    /**
     * 消费消息的转换配置.
     *
     * @return
     */
    @Bean
    public MessagingMessageConverter simpleMapperConverter() {
        MessagingMessageConverter messagingMessageConverter = new MessagingMessageConverter();
        messagingMessageConverter.setHeaderMapper(new SimpleKafkaHeaderMapper());
        // 配置JSON转换器
        MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
        messagingMessageConverter.setMessagingConverter(converter);
        return messagingMessageConverter;
    }


    /**
     * 配置消费者工厂
     */
    @Bean
    public ConsumerFactory consumerFactory() {
        DefaultKafkaConsumerFactory factory = new DefaultKafkaConsumerFactory(consumerConfigs(), () -> new StringDeserializer(), () -> new StringDeserializer());
        return factory;
    }

    private Map consumerConfigs() {
        Map map = new HashMap();
        // 生产者连接服务器的地址。
        map.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.32.128:9092");
        // 关闭消费者的自动提交
        map.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        // 当无法反序列化时，就会抛出异常，所以需要捕获
//        map.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
//        map.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
//        // ErrorHandlingDeserializer代理的是key->”字符串序列化“,value->json序列化。
//        map.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, StringSerializer.class);
//        map.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class.getName());


        return map;
    }

    /**
     * 监听器容器，管理并调用监听器。
     *
     * @return
     */
//    @Bean
    public ConcurrentMessageListenerContainer container(ConsumerFactory c) {
        ContainerProperties properties = new ContainerProperties("one", "tow");
        properties.setGroupId(UUID.randomUUID().toString());
        // 设置日志级别
//        properties.setCommitLogLevel(LogIfLevelEnabled.Level.DEBUG);
        // 当kafka中没有任何的topic时。禁止容器启动。
//        properties.setMissingTopicsFatal(false);
        // 当kafka抛出认证异常的时候，消费允许重试的时间。默认是禁止的。这会导致容器停止
//        properties.setAuthExceptionRetryInterval();
        // 设置提交方式
//        properties.setAckMode();
        // 使用自定义的线程池,默认使用SimpleAsyncTaskExecutor线程池
//        properties.setConsumerTaskExecutor();
        // 设置消息处理的监听器
        properties.setMessageListener(new MessageListener<String, Object>() {
            /**
             * 会导致覆盖@KafkaListener注解监听
             * @param data the data to be processed.
             */
            @Override
            public void onMessage(ConsumerRecord<String, Object> data) {
                System.out.println("监听器接收到消息。。。。。。。。。");
                System.out.println(data.value());
                Headers headers = data.headers();
                for (Header header : headers) {
                    System.out.println(header.key() + "-" + header.value());
                }
            }

            /**
             * 如果开启了手动提交，那么就必须自己主动提交偏移量。
             * @param data 接收到的数据
             * @param acknowledgment 消息回复
             * @param consumer 消费者
             */
//            @Override
//            public void onMessage(ConsumerRecord<String, Object> data, Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
//
//            }
        });
        ConcurrentMessageListenerContainer<Object, Object> container =
                new ConcurrentMessageListenerContainer<>(c, properties);

        // 允许并发情况下，创建3个KafkaMessageListenerContainer容器实列。
//        container.setConcurrency(3);
        // 配置拦截器
        container.setRecordInterceptor(record -> {
            System.out.println("拦截器被执行........................");
            return record;
        });
        // 启动该容器。
//        container.start();
        return container;
    }

    /**
     * 创建监听器工厂，工厂是专门创建消息监听器容器的。
     *
     * @return
     */
//    @Bean
    public KafkaListenerContainerFactory containerFactory() {
        ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        // 容器的自定义配置处理,可以在创建容器的时候对配置进行修改。
        factory.setContainerCustomizer((container) -> {
        });
        // 消息的转换
//        factory.setMessageConverter();
        return factory;
    }

    /**
     * 使用它可以设置kafka的自定义方法参数解析
     *
     * @param registrar the registrar to be configured
     */
    @Override
    public void configureKafkaListeners(KafkaListenerEndpointRegistrar registrar) {
        // 注册自定义的方法参数解析
        registrar.setCustomMethodArgumentResolvers(new HandlerMethodArgumentResolver() {
            @Override
            public boolean supportsParameter(MethodParameter parameter) {
                // 方法参数是否是User接收
                return User.class.isAssignableFrom(parameter.getParameterType());
            }

            @Override
            public Object resolveArgument(MethodParameter parameter, Message<?> message) throws Exception {
                Object payload = message.getPayload();
                System.out.println(payload);
                return new User("1", 25);
            }
        });
    }
}
